package b.acct.impl

import akka.Done
import com.datastax.driver.core.{BoundStatement, PreparedStatement}
import com.lightbend.lagom.scaladsl.persistence.ReadSideProcessor.ReadSideHandler
import com.lightbend.lagom.scaladsl.persistence.cassandra.{CassandraReadSide, CassandraSession}
import com.lightbend.lagom.scaladsl.persistence.{AggregateEventTag, EventStreamElement, ReadSideProcessor}

import scala.collection.immutable.Seq
import scala.concurrent.{ExecutionContext, Future}

class C账户ReadSide(
                readSide: CassandraReadSide,
                session: CassandraSession)(implicit ec: ExecutionContext) extends ReadSideProcessor[T账户Event]{

  @volatile private var v账户Create: PreparedStatement = _ // initialized in prepare
  @volatile private var v账户Update: PreparedStatement = _ // initialized in prepare
  @volatile private var v转账Create: PreparedStatement = _ // initialized in prepare
  @volatile private var v转账Update: PreparedStatement = _ // initialized in prepare

  def fDDLs(): Future[Done] = {
    val account = s"""
                     |CREATE TABLE IF NOT EXISTS account (
                     |  no ascii,
                     |  balance decimal,
                     |  status smallint,
                     |  PRIMARY KEY (no)
                     |)""".stripMargin
    val transfer = s"""
                      |CREATE TABLE IF NOT EXISTS transfer (
                      |  sequence bigint,
                      |  no ascii,
                      |  peerNo ascii,
                      |  amount decimal,
                      |  status smallint,
                      |  PRIMARY KEY (sequence, no)
                      |)""".stripMargin
    Future.sequence(Seq[Future[Done]](
      session.executeCreateTable(account),
      session.executeCreateTable(transfer)
    )).map(_ => Done)
  }

  def fSQLs: Future[Done] = {
    Future.sequence(Seq[Future[Done]](
      session.prepare("INSERT INTO account (no, balance, status) VALUES (?, 0.00, 0)")
        .map { ps => v账户Create = ps; Done},
      session.prepare("UPDATE account SET balance = ? WHERE no = ?")
        .map { ps => v账户Update = ps; Done},
      session.prepare("INSERT INTO transfer (sequence, no, peerNo, amount, status) VALUES (?, ?, ?, ?, 0)")
        .map { ps => v转账Create = ps; Done},
      session.prepare("UPDATE transfer SET status = ? WHERE sequence = ? AND no = ?")
        .map { ps => v转账Update = ps; Done}
    )).map(_ => Done)
  }

  private def f账户Create(event: CC新账户Event): Future[Seq[BoundStatement]] = {
    println(s"f账户Create: $event")
    Future.successful({
      val stmt = v账户Create.bind()
      stmt.setString("no", event.v账户.v账号)
      Seq(stmt)
    })
  }

  private def f转账Create(event: CC账户余额预留Event): Future[Seq[BoundStatement]] = {
    println(s"f转账Create: $event")
    Future.successful({
      val stmt = v转账Create.bind()
      stmt.setLong("sequence", event.v事务ID)
      stmt.setString("no", event.v账号)
      stmt.setString("peerNo", event.v对方账号)
      stmt.setDecimal("amount", event.v金额.bigDecimal)
      Seq(stmt)
    })
  }

  private def f账户Update(event: CC账户新余额Event): Future[Seq[BoundStatement]] = {
    println(s"f账户Update: $event")
    Future.successful({
      val stmt = v账户Update.bind()
      stmt.setDecimal(0, event.v新余额.bigDecimal)
      stmt.setString(1, event.v账号)
      val stmtOut = v转账Update.bind()
      stmtOut.setShort("status", 1.toShort)
      stmtOut.setLong("sequence", event.v事务ID)
      stmtOut.setString("no", event.v账号)
      Seq(stmt, stmtOut)
    })
  }

  private def f转账Update(event: CC账户余额恢复Event): Future[Seq[BoundStatement]] = {
    println(s"f转账Update: $event")
    Future.successful({
      val stmtOut = v转账Update.bind()
      stmtOut.setShort("status", (-1).toShort)
      stmtOut.setLong("sequence", event.v事务ID)
      stmtOut.setString("no", event.v账号)
      Seq(stmtOut)
    })
  }

  override def aggregateTags(): Set[AggregateEventTag[T账户Event]] = {
    Set(O账户Event.TAG)
  }

  override def buildHandler(): ReadSideHandler[T账户Event] = {
    val b = readSide.builder[T账户Event]("ACCT")
    b.setGlobalPrepare(fDDLs)
    b.setPrepare(tag => fSQLs)
    b.setEventHandler((c: EventStreamElement[CC新账户Event]) => f账户Create(c.event))
    b.setEventHandler((c: EventStreamElement[CC账户余额预留Event]) => f转账Create(c.event))
    b.setEventHandler((c: EventStreamElement[CC账户新余额Event]) => f账户Update(c.event))
    b.setEventHandler((c: EventStreamElement[CC账户余额恢复Event]) => f转账Update(c.event))
    b.build()
  }
}

